package org.example.mq.mqserver;

import org.example.mq.common.*;
import org.example.mq.common.network.mqservice.BasicArguments;
import org.example.mq.common.network.mqservice.BasicReturns;
import org.example.mq.common.network.mqservice.Exchange.ExchangeDeclareArguments;
import org.example.mq.common.network.mqservice.Exchange.ExchangeDeleteArguments;
import org.example.mq.common.network.mqservice.MSGQueue.QueueDeclareArguments;
import org.example.mq.common.network.mqservice.MSGQueue.QueueDeleteArguments;
import org.example.mq.common.network.mqservice.Message.BasicAckArguments;
import org.example.mq.common.network.mqservice.Message.BasicConsumeArguments;
import org.example.mq.common.network.mqservice.Message.BasicPublicArguments;
import org.example.mq.common.network.mqservice.QueueBind.QueueBindArguments;
import org.example.mq.common.network.mqservice.QueueBind.QueueUnBindArguments;
import org.example.mq.common.network.mqservice.SubScribeReturns;
import org.example.mq.mqserver.core.BasicProperties;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BrokerServer {
    private ServerSocket serverSocket = null;


    //在一个brokerServer中只存在一个虚拟主机
    private VirtualHost virtualHost = new VirtualHost("default");

    //key是channelId，value对应的Socket对象（一个虚拟机内，存在多个socket对象）
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();

    //引入一个线程池，来处理多个客户端的请求
    private ExecutorService executorService = null;

    //控制服务器是否继续运行
    private volatile boolean runnable = true;

    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }

    public void start() throws IOException {
        System.out.println("启动");
        //这里动态创建线程池，需要几个就启动几次
        executorService = Executors.newCachedThreadPool();
        try {
            while (runnable) {
                Socket clientSocket = serverSocket.accept();
                executorService.submit(() -> {
                    try {
                        processConnection(clientSocket);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                });
            }
        }catch (SocketException e){
            System.out.println("服务器停止运行");
        }
    }

    public void stop() throws IOException {
//        将线程工作的循环停止
        runnable = false;
//        将线程池中的任务都放弃，让线程都销毁
        executorService.shutdownNow();
//        关闭连接
        serverSocket.close();
    }

    //通过方法，处理一个客户端的连接
    private void processConnection(Socket clientSocket) throws IOException {
        try (InputStream inputStream = clientSocket.getInputStream();
             OutputStream outputStream = clientSocket.getOutputStream()) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                while (true) {
//                    1.读取请求并解析
                    Request request = readRequest(dataInputStream);
//                    2.根据请求计算响应
                    Response response = process(request, clientSocket);
//                    3.把响应写会客户端
                    writeResponse(dataOutputStream, response);
                }
            } catch (EOFException | SocketException e) {
//                借助异常结束循环
                System.out.println("连接关闭");
            } catch (MqException e) {
                throw new RuntimeException(e);
            }
        } catch (IOException | ClassNotFoundException e) {
            System.out.println("连接出现异常");
            e.printStackTrace();
        } finally {
            clientSocket.close();
            clearClosedSession(clientSocket);
        }
    }


    /*
     * 把响应写会客户端
     * */
    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        dataOutputStream.flush();
    }

    /*
    * 读取请求并解析（根据约定好的格式进行读取（类型+长度+内容））
    * */
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());
        byte[] payLoad = new byte[request.getLength()];
        int n = dataInputStream.read(payLoad);
        if(n!=request.getLength()){
            throw new IOException("读取请求格式错误");
        }
        request.setPayload(payLoad);
        return request;
    }

    /*
     * 根据请求计算响应
     * */
    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
//        1. payload的内容根据type来决定,计算响应，必须先对请求进行更近一步的解析（将去取的内容全部转换为BasicArguments（读取到的内容全是BasicArguments的子类））
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
        System.out.println("rid = " + basicArguments.getRid() + ",channelId= " + basicArguments.getChannelId());
//        2. 根据type的值，进行区分请求要干什么
        boolean ok = true;
        if (request.getType() == 0x1) {
            //创建channel
            sessions.put(basicArguments.getChannelId(), clientSocket);
        } else if (request.getType() == 0x2) {
            //删除channel
            sessions.remove(basicArguments.getChannelId());
        } else if (request.getType() == 0x3) {
            ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
            ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getExchangeType(), arguments.isDurable(), arguments.isAutoDelete(), arguments.getArguments());
        } else if (request.getType() == 0x4) {
            ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
            ok = virtualHost.exchangeDelete(arguments.getExchangeName());
        } else if (request.getType() == 0x5) {
            QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
            ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable(), arguments.isExclusive(),
                    arguments.isAutoDelete(), arguments.getArguments());
        } else if (request.getType() == 0x6) {
            QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
            ok = virtualHost.queueDelete(arguments.getQueueName());
        } else if (request.getType() == 0x7) {
            QueueBindArguments arguments = (QueueBindArguments) basicArguments;
            ok = virtualHost.queueBind(arguments.getQueueName(),arguments.getExchangeName(),arguments.getBindingKey());
        } else if (request.getType() == 0x8) {
            QueueUnBindArguments arguments = (QueueUnBindArguments) basicArguments;
            ok = virtualHost.queueUnbind(arguments.getQueueName(),arguments.getExchangeName());
        } else if (request.getType() == 0x9) {
            BasicPublicArguments arguments = (BasicPublicArguments) basicArguments;
            ok = virtualHost.basicPublic(arguments.getExchangeName(),arguments.getRoutingKey(),arguments.getBasicProperties(),
                    arguments.getBody());
        } else if (request.getType() == 0xa) {
            //实现订阅消息
            BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
            ok = virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(),
                    new Consumer() {
                        //回调函数要做的事：将消息推送给客户端
                        @Override
                        public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                            //这里consumerTag就是对应的channelId，根据channelId可以得到socket对象，从而发送给对应的客户端（返回给socket对象）
//                            1.根据channelId查找socket对象(consumerTag的值应该和channelId的值一样)
                            Socket clientSocket = sessions.get(consumerTag);
                            if(clientSocket == null||clientSocket.isClosed()){
                                throw new MqException("订阅消息的客户端已经关闭");
                            }
//                            2.构造响应数据
                            //返回的具体数据/描述
                            SubScribeReturns subScribeReturns = new SubScribeReturns();
                            //只有一个请求对应一个响应的过程，才需要rid匹配
                            subScribeReturns.setRid("");
                            subScribeReturns.setChannelId(consumerTag);
                            subScribeReturns.setOk(true);
                            subScribeReturns.setConsumerTag(consumerTag);
                            subScribeReturns.setBasicProperties(basicProperties);
                            subScribeReturns.setBody(body);
                            //将数据转为2进制，然后写入Respond中并返回
                            byte[] payload = BinaryTool.toBytes(subScribeReturns);

                            Response response = new Response();
                            response.setType(0xc);
                            response.setLength(payload.length);
                            response.setPayload(payload);

//                            3，将数据写回给客户端
                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            writeResponse(dataOutputStream,response);
                        }
                    });
        } else if (request.getType() == 0xb) {
            BasicAckArguments arguments = (BasicAckArguments) basicArguments;
            ok = virtualHost.basicAck(arguments.getQueueName(),arguments.getMessageId());
        }else {
            throw new MqException("未知的type");
        }
//        3.构造响应
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setOk(ok);
        byte[] payLoad = BinaryTool.toBytes(basicReturns);

        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payLoad.length);
        response.setPayload(payLoad);
        return response;
    }


    /*
    * 遍历SessionHash表，把该关闭的socket对应的键值对都删掉
    * 比如：aaa-socket1  bbb-socket1  ccc-socket1
    * 删除 aaa也就是删除socket1，但是bbb，ccc中还在关联，所以把和socket1关联的键值对都删除
    * */
    private void clearClosedSession(Socket clientSocket) {
        List<String> toDeleteChannelId = new ArrayList<>();
//        统计要删除的
        for(Map.Entry<String,Socket> entry : sessions.entrySet()){
            if(entry.getValue()==clientSocket){
                //集合类大忌：不能一遍删除一遍遍历
                toDeleteChannelId.add(entry.getKey());
            }
        }
//        开始删除
        for (String channelId : toDeleteChannelId){
            sessions.remove(channelId);
        }
    }

}
